import asyncio
import tarfile
import sqlite3
from concurrent.futures import ThreadPoolExecutor


# 假设数据库连接和游标
conn = sqlite3.connect('example.db')
cur = conn.cursor()

# 处理单个文件的异步任务
async def process_file(tar, file_name, tar_loc, id, executor):
    await asyncio.get_event_loop().run_in_executor(executor, handle_file, tar, file_name, tar_loc, id)

# 处理文件的同步操作
def handle_file(tar, file_name, tar_loc, id):
    try:
        # 模拟处理文件
        print(f"处理文件: {file_name} (在 tar 包 {tar_loc} 中)")
        
        # 执行数据库插入（假设数据处理完成）
        cur.execute("INSERT OR IGNORE INTO tar (file_name, tar_loc, id) VALUES (?,?,?)", 
                    (file_name, tar_loc, id))
        conn.commit()
    except Exception as e:
        print(f"处理文件 {file_name} 时发生错误: {e}")

# 处理 tar 包的异步任务
async def process_tarfile(tar_loc, id, executor):
    try:
        with tarfile.open(tar_loc, 'r') as tar:
            file_names = tar.getnames()

        # 对 tar 包中的每个文件进行并发处理
        tasks = [process_file(tar, file_name, tar_loc, id, executor) for file_name in file_names]
        await asyncio.gather(*tasks)
    except Exception as e:
        print(f"处理 tar 包 {tar_loc} 时发生错误: {e}")

# 异步任务：处理队列中的 tar 包
async def process_tar_queue(queue, executor):
    while True:
        tar_loc, id = await queue.get()
        if tar_loc is None:
            # None 用于终止任务
            break
        await process_tarfile(tar_loc, id, executor)
        queue.task_done()

# 异步主程序
async def main():
    # 创建一个异步队列
    queue = asyncio.Queue()
    executor = ThreadPoolExecutor(max_workers=4)  # 创建线程池

    # 启动消费者任务（处理 tar 包）
    consumer = asyncio.create_task(process_tar_queue(queue, executor))

    # 模拟传入 tar 包的过程
    tar_files = [
        ("path/to/tarfile1.tar", "id1"),
        ("path/to/tarfile2.tar", "id2"),
        ("path/to/tarfile3.tar", "id3"),
    ]
    
    # 将 tar 包任务放入队列
    for tar_loc, id in tar_files:
        await queue.put((tar_loc, id))

    # 等待队列处理完成
    await queue.join()

    # 停止消费者任务
    await queue.put((None, None))  # 终止消费者
    await consumer

# 启动事件循环并执行异步任务
if __name__ == "__main__":
    asyncio.run(main())

# 关闭数据库连接
cur.close()
conn.close()
